/**  
 * @Title: ExecutorSerice.java
 * @Description: TODO(描述)
 * @author liuyi
 * @date 2023-08-16 08:00:33 
 */
package com.fsxgt.datagrid.console.service.impl;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpHeaders;
import org.springframework.stereotype.Service;

import com.fsxgt.datagrid.console.dao.FlowExecLogMapper;
import com.fsxgt.datagrid.console.exception.RunningException;
import com.fsxgt.datagrid.core.component.HttpComponent;
import com.fsxgt.datagrid.core.utils.ReturnT;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * @ClassName: ExecutorSerice
 * @Description: TODO(描述)
 * @author liuyi
 * @date 2023-08-16 08:00:33
 */
@Service
@Slf4j
public class ExecutorService {
	@Autowired
	private FlowExecLogMapper flowExecLogMapper;
	@Autowired
	private HttpComponent httpComponent;

	private static Map<String, RuningTask> threadMap = new ConcurrentHashMap<String, RuningTask>();
	
	public interface Callback{
		public void handle(String overTime, String startTime, String status, String msg, String logPath, String execId) throws Exception;
	}

	@Data
	private class RuningTask {

		private String server;
		private Date triggerTime;
		private Callback callback;

		public RuningTask(String server, Date triggerTime,Callback callback) {
			this.server = server;
			this.triggerTime = triggerTime;
			this.callback = callback;
		}
	}
	
	public ReturnT<Map> start(String objId,String objType, String server, String execId, Date triggerTime, Map context) throws RunningException{
		return start( objId,objType,  server,  execId,  triggerTime,  null,null,null,null, context,null);
	}

	public ReturnT<Map> start(String objId,String objType, String server, String execId, Date triggerTime, String version,Integer threads,Integer timeout,Integer retryCount, Map context,Callback callback)
			throws RunningException {

		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

		String triggerTimeStr = format.format(triggerTime);
		
		String obj=objId+"#"+objType;

		synchronized (this) {
			if (threadMap.containsKey(obj)) {
				if (timeout!=null&&threadMap.get(obj).getTriggerTime().getTime() + timeout*1000 > System.currentTimeMillis()) {
					throw new RunningException("调度已经在运行-"+obj);
				}
			} 
			threadMap.put(obj, new RuningTask(server, triggerTime,callback));
		}
		try {
			HttpHeaders headers = new HttpHeaders();
			headers.add("Content-Type", "application/json; charset=UTF-8");
			Map param = new HashMap();
			param.put("execId", execId);
			param.put("objId", objId);
			param.put("objType", objType);
			param.put("triggerTime", triggerTimeStr);
			param.put("version", version);
			param.put("threads", threads);
			
			param.put("context", context);
			
			param.put("timeout", timeout!=null?timeout*1000:null);
			param.put("retryCount", retryCount);

            String url="http://" + server + "/executor/start";
            
			ReturnT<Map> result = httpComponent.sendHttpPostRetry(url, param,
					ReturnT.class);
			
			log.info(url+"-"+param);

			result.setData(new HashMap());
			result.getData().put("server", server);
			result.getData().put("execId", execId);
			result.getData().put("objId", objId);
			result.getData().put("objType", objType);
			result.getData().put("triggerTime", triggerTimeStr);
			
			
			

			if (!result.getCode().equals("200")) {
				threadMap.remove(obj);
			}
			return result;
		} catch (Exception e) {
			threadMap.remove(obj);
			throw e;
		}

	}

	public ReturnT stop(String execId, String server) {

		Map param = new HashMap();
		param.put("execId", execId);

		return httpComponent.sendHttpPostRetry("http://" + server + "/executor/stop", param, ReturnT.class);

	}

	public ReturnT getLog(String execId, String server, String triggerTime) {
		// TODO Auto-generated method stub

		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

		Map param = new HashMap();
		param.put("execId", execId);
		param.put("triggerTime", triggerTime);

		return httpComponent.sendHttpPostRetry("http://" + server + "/executor/getLog", param, ReturnT.class);

	}

	public void callback(String overTime, String startTime, String status, String msg, String logPath, String execId,
			String objId,String objType) throws Exception {

		Callback callback = null;
		
		String obj=objId+"#"+objType;

		synchronized (this) {
			if (threadMap.containsKey(obj)) {
				callback = threadMap.get(obj).getCallback();
				threadMap.remove(obj);
			}
		}
		

		if(callback!=null) {
			callback.handle( overTime,  startTime,  status,  msg,  logPath,  execId);
		}
		

	}

}
